package edu.wisc.hevea.topic;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import edu.wisc.hevea.constants.ApplicationConstants;
import edu.wisc.hevea.hsql.Message;
import edu.wisc.hevea.hsql.MessageDAO;

public class TopicHandler {
	
	private static Log log = LogFactory.getLog(TopicHandler.class);
	private static int MAX_QUEUE_SIZE = ApplicationConstants.HIGH_WATER_MARK;
	
	private String topic;
	private ConcurrentLinkedQueue<Message> msgQueue;
	
	public ConcurrentLinkedQueue<Message> getMsgQueue() {
		return msgQueue;
	}

	public void setMsgQueue(ConcurrentLinkedQueue<Message> msgQueue) {
		this.msgQueue = msgQueue;
	}
	
	private String hostName;

	private MessageDAO messageDAO;

	public TopicHandler(String topic, String hostName) {
		this.topic = topic;
		this.msgQueue = new ConcurrentLinkedQueue<Message>();
		this.messageDAO = MessageDAO.getMessageDAO(hostName);
		this.hostName = hostName;
	}

	public Message putMsgInPending(String content) throws SQLException {
		Message m = new Message();

		String msgId = String.format("%s_%s", hostName, new Date().getTime());
		m.setMsgId(msgId);
		m.setContent(content);
		m.setTopic(topic);
		m.setContent(content);
		m.setStatus("PENDING");

		messageDAO.insertMessage(MessageDAO.MY_MSG_TABLE_NAME, m);
		
		if(msgQueue.size() < MAX_QUEUE_SIZE) {
			log.debug("Adding message to queue " + m.getMsgId());
			msgQueue.add(m);
		}

		return m;
	}
	
	public synchronized Message getMessageForConsumption() {
		Message m;
		if (msgQueue.size() == 0) {
			m = new Message();
			m.setContent("NO_MSG_AVAILABLE");
			m.setMsgId("-1");
			m.setTopic(topic);
		} else {
			m = msgQueue.remove();
			log.info("Removing message from Q!" + m.getMsgId());
		}
		return m;
	}

	public void putMsgInConsumed(Message m) throws SQLException {
		// msgQueue.remove(m);
		if (m.getMsgId().contains(hostName)) {
			messageDAO.updateMessageToConsumed(MessageDAO.MY_MSG_TABLE_NAME, m);
		} else {
			messageDAO.updateMessageToConsumed(
					MessageDAO.REPLICATED_MSG_TABLE_NAME, m);
		}
	}
	
	public void putReconciledMsgs(List<String> msgIds) {
		for(String msgId : msgIds) {
			try {
				messageDAO.updateReplicatedMessageToConsumed(msgId);
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}

	public Message putMsgInReplicatedTable(Message m) throws SQLException {
		messageDAO.insertMessage(MessageDAO.REPLICATED_MSG_TABLE_NAME, m);
		return m;
	}
	
	public void putPendingMsgsInQueue() throws SQLException {
		if (msgQueue.size() >= MAX_QUEUE_SIZE) {
			log.info("Queue already full!");
			return;
		}

		int numMsgsToAdd = MAX_QUEUE_SIZE - msgQueue.size();

		List<Message> pendingMsgs = messageDAO
				.getAllPendingMessages(numMsgsToAdd);
		
		
		for(Message m : pendingMsgs) {
			if(msgQueue.size() >= MAX_QUEUE_SIZE) {
				break;
			}
			
			if(!msgQueue.contains(m)) {
				msgQueue.add(m);	
			}			
		}				
	}
	
	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append("Topic: " + topic);
		sb.append("\nMsgs: ");
		for(Message m : msgQueue) {
			sb.append(m.toString() + "\n");
		}
		return sb.toString();		
	}
}
